package org.atguigu.gmall.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.SQLUtil;

/**
 * @Author zenghaifeng
 * @Date 2023/4/24 21:13
 * @Package: org.atguigu.gmall.realtime.app.dwd.db
 * @Class: Dwd_02_DwdInteractionCommentInfo
 * @Description: TODO
 * @Version 1.0
 */
public class Dwd_02_DwdInteractionCommentInfo extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdInteractionCommentInfo().init(30002,2,"Dwd_02_DwdInteractionCommentInfo");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        readOdsDb(tbe,"Dwd_02_DwdInteractionCommentInfo");

        // filter comment rows
        Table commentInfo = tbe.sqlQuery("select " +
                " `data`['id'] id," +
                " `data`['user_id'] user_id," +
                " `data`['sku_id'] sku_id," +
                " `data`['appraise'] appraise," +
                " `data`['comment_txt'] comment_txt," +
                " ts,  " +
                " pt " +
                "from ods_db " +
                "where `database`='gmall2023' " +
                "and `table`='comment_info' " +
                "and `type`='insert' ");
        // register temporary table name
        tbe.createTemporaryView("comment_info",commentInfo);
        // read base_dic
        readBaseDic(tbe);
        // join
        String joinSql = " SELECT " +
                " id, " +
                " user_id, " +
                " sku_id, " +
                " appraise AS appraise_code, " +
                " dic.dic_name AS appraise_name, " +
                " comment_txt, " +
                " ts " +
                " FROM `comment_info` AS `ci` " +
                " JOIN `base_dic` FOR SYSTEM_TIME AS OF `ci`.pt AS `dic` " +
                " ON `ci`.appraise = `dic`.dic_code ";
        Table result = tbe.sqlQuery(joinSql);

        // write into kafka
        tbe.executeSql(" create table dwd_interaction_comment_info(" +
                " id string, " +
                " user_id string, " +
                " sku_id string, " +
                " appraise_code string, " +
                " appraise_name string, " +
                " comment_txt string, " +
                " ts bigint " +
                ") " + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO));
        result.executeInsert("dwd_interaction_comment_info");
    }
}
